package com.gizwits.AkkaActor

import java.util.Properties
import scala.collection.JavaConversions._
import akka.actor.{Props, Actor}
import akka.actor.Actor._
import akka.event.Logging
import com.gizwits.util.CaseClasssDomain._
import com.gizwits.util.Constants
import kafka.producer.{Producer, ProducerConfig}

class ProductMasterActor extends  Actor{
  val props = new Properties()
  val log = Logging.getLogger(context.system, this)
  import context._
  override def preStart(): Unit = {
    super.preStart()
    //do pre something
    props ++= Map(
      "serializer.class" -> Constants.SERIALIZEENCODERRCLASS,
      "metadata.broker.list" -> Constants.BROKERS,
      "key.serializer.class" ->"kafka.serializer.StringEncoder",
      "producer.type"-> "async",
      "compression.codec" -> "snappy"
    )
  }
  override def receive: Receive = {

    case  Calu(num)=>{
      val config = new ProducerConfig(props)
      val producer = new Producer[String, User](config)
      val p2 = context.actorOf(Props(new ProductWorkerActor(producer)), name = "p2")

          for(u<-genMsg(num))
          {
            log.info(u.toString)
            p2 ! u
          }
        //context.system.shutdown()

    }
    case _=>{
          log.info("no macth")
    }
  }
    //test genter info data
   def genMsg(num:Int)={

     val  ulist= collection.mutable.ListBuffer[User]()

       for(i <-1 to  num){
            ulist.+=:(User("feel"+1,10+i))
       }
     ulist

   }
  override def postStop(): Unit = {
    super.postStop()
    //stop somthing
     // context.stop(self)
  }
}
